use std::sync::Arc;

use ecksport_rpc::{errors::RpcError, prelude::*};
use futures::TryFutureExt;
use tracing::*;

use aspect_codec::ecksport::AsAspCodec;
use aspect_core::message_queue::{D2DChannelSpec, MessageQueueId};
use aspect_proto::{errors::AspProtoError, msg_queue};

use crate::storage::HomeserverStorage;

pub struct DeviceMqServer {
    storage: Arc<HomeserverStorage>,
}

impl DeviceMqServer {
    pub fn new(storage: Arc<HomeserverStorage>) -> Arc<Self> {
        Arc::new(Self { storage })
    }
}

impl aspect_proto::core::MqServ for DeviceMqServer {
    fn mq_list(
        &self,
        _pdata: PeerData,
        req: AsAspCodec<msg_queue::MqListRequest>,
    ) -> impl std::future::Future<Output = RpcResult<AsAspCodec<msg_queue::MqListResponse>>> + Send
    {
        async move {
            let msg_queue::MqListRequest {
                creds,
                device_ident,
            } = req.into_inner();

            // FIXME switch to ecksport based auth
            if self
                .storage
                .check_device_login(&creds.username, &creds.password, &device_ident)
                .await
                .is_err()
            {
                return Err(AspProtoError::InvalidPassword(()).into());
            }

            let queue_statuses = self
                .storage
                .get_device_queues(&device_ident)
                .map_err(|e| RpcError::server(10000, e))
                .await?;

            let resp = msg_queue::MqListResponse {
                queues: queue_statuses
                    .into_iter()
                    .map(|qs| msg_queue::QueueViewState {
                        id: qs.id,
                        device_pos: qs.reader_pos.unwrap_or_default(),
                        // TODO should we fix this inconsistency?
                        queue_pos: if qs.next_message_pos > 0 {
                            qs.next_message_pos
                        } else {
                            u64::MAX
                        },
                    })
                    .collect(),
            };
            Ok(AsAspCodec::new(resp))
        }
    }

    fn mq_fetch(
        &self,
        _pdata: PeerData,
        range: AsAspCodec<msg_queue::MqFetchRange>,
    ) -> impl std::future::Future<Output = RpcResult<AsAspCodec<msg_queue::MqEntries>>> + Send {
        let msg_queue::MqFetchRange {
            creds,
            device_id,
            queue_id,
            start_idx,
            end_idx,
        } = range.into_inner();

        async move {
            // FIXME switch to ecksport based auth
            if self
                .storage
                .check_device_login(&creds.username, &creds.password, &device_id)
                .await
                .is_err()
            {
                return Err(AspProtoError::InvalidPassword(()).into());
            }

            let next_idx = self
                .storage
                .get_device_queue_next_idx(&device_id, &queue_id)
                .map_err(|e| RpcError::server(10000, e))
                .await?;

            let msgs = self
                .storage
                .query_device_queue_msgs(&device_id, &queue_id, start_idx, end_idx)
                .map_err(|e| RpcError::server(10000, e))
                .await?;

            let resp = msg_queue::MqEntries {
                cur_queue_pos: next_idx,
                entries: msgs,
            };

            Ok(resp.into())
        }
    }

    fn mq_consume(
        &self,
        _pdata: PeerData,
        _req: AsAspCodec<msg_queue::ConsumeRequest>,
    ) -> impl std::future::Future<Output = RpcResult<()>> + Send {
        async {
            // not yet implemented, but also we don't want to panic here
            warn!("ignoring mq_consume request");
            Ok(())
        }
    }

    fn mq_send_d2d(
        &self,
        _pdata: PeerData,
        req: AsAspCodec<msg_queue::MqDirectDeviceSendRequest>,
    ) -> impl std::future::Future<Output = RpcResult<()>> + Send {
        let msg_queue::MqDirectDeviceSendRequest {
            creds,
            device_id,
            target_device_id,
            payload,
        } = req.into_inner();

        async move {
            // FIXME switch to ecksport based auth
            if self
                .storage
                .check_device_login(&creds.username, &creds.password, &device_id)
                .await
                .is_err()
            {
                return Err(AspProtoError::InvalidPassword(()).into());
            }

            let len = payload.len();
            trace!(%device_id, %target_device_id, %len, "relaying D2D message");

            let mqid = MessageQueueId::D2D(D2DChannelSpec::new(device_id, target_device_id));
            self.storage
                .create_queue_ifne(&mqid)
                .map_err(|e| RpcError::server(10000, e))
                .await?;

            trace!(%device_id, %mqid, "created queue");

            // TODO when this has cross-homeserver messages this will have to be
            // much more sophisticated and go through some router system
            self.storage
                .queue_device_msg(&mqid, &payload)
                .map_err(|e| RpcError::server(10000, e))
                .await?;
            trace!(%device_id, %target_device_id, %len, "relayed device message");

            Ok(())
        }
    }
}
